package personal.wang.worker;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.FixedRecvByteBufAllocator;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import personal.wang.config.DataMockYamlReader;
import personal.wang.constant.ConstantVar;
import personal.wang.domain.HeartbeatBean;
import personal.wang.enm.MessageType;
import personal.wang.handler.WorkerHandler;
import personal.wang.manager.NetworkManager;
import personal.wang.manager.ZookeeperManager;
import personal.wang.message.*;
import personal.wang.util.ParameterUtil;
import personal.wang.util.SystemInfoUtil;
import personal.wang.zookeeper.HAUtils;

import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.Properties;
import java.util.UUID;

/**
 * @className: Worker
 * @Description:
 * @Author: wangyifei
 * @Date: 2025/2/14 10:55
 */
public class Worker {
    private static final Logger logger = LoggerFactory.getLogger(Worker.class);
    public static void main(String[] args) throws InterruptedException {

//        Bootstrap bootstrap = new Bootstrap();
//        NioEventLoopGroup group = new NioEventLoopGroup();
        Properties properties = ParameterUtil.fromArgument(args);
        String zkQuorum = DataMockYamlReader.getValueAsString(properties.getProperty("path") , ConstantVar.ZOOKEEPER_CONNECTION);

        String zookeeperHaPath = DataMockYamlReader.getValueAsString(ConstantVar.ZOOKEEPER_HA_PATH);
        CuratorFramework curatorFramework = HAUtils.buildClient();
        curatorFramework.start();
        try {
            byte[] bytes = curatorFramework.getData().forPath(zookeeperHaPath);
            String s = StringUtils.toEncodedString(bytes, StandardCharsets.UTF_8);
            String[] split = s.split(":");
            NetworkManager.reconnectBoss(split[0] , split[1]);
        } catch (Exception e) {
            e.printStackTrace();
        }
        curatorFramework.close();

        Channel client = NetworkManager.getClient();
        ZookeeperManager zookeeperManager = new ZookeeperManager(null, null);
        zookeeperManager.startWorker();
//        Channel client = bootstrap.group(group)
//                .channel(NioSocketChannel.class)
//                .handler(new ChannelInitializer<NioSocketChannel>() {
//                    @Override
//                    protected void initChannel(NioSocketChannel ch) throws Exception {
//                        ch.pipeline().addLast(new MessageEncoder());
//                        ch.pipeline().addLast(new MessageDecoder());
//                        ch.pipeline().addLast(new WorkerHandler());
//                    }
//                }).option(ChannelOption.RCVBUF_ALLOCATOR , new FixedRecvByteBufAllocator(10 * 1024))
//                .option(ChannelOption.SO_RCVBUF , 10*1024)
//                .connect(new InetSocketAddress("localhost", 8082))
//                .sync().channel();
        HeartbeatThread heartbeatThread = new HeartbeatThread(()->{
            HeartbeatBean.Endpoint endpoint = HeartbeatBean.Endpoint.newBuilder()
                    .setHost(SystemInfoUtil.getLocalHost())
                    .setIp(SystemInfoUtil.getLocalIP())
                    .build();
            HeartbeatBean.Report report = HeartbeatBean.Report.newBuilder()
                    .setModel("")
                    .setMessageCnt(0D)
                    .build();
            HeartbeatBean.Heartbeat heartbeat = HeartbeatBean.Heartbeat.
                    newBuilder()
                    .setEndpoint(endpoint)
                    .setReport(report)
                    .build();
            Header header = new Header(1 , 36
                    , heartbeat.getSerializedSize()
                    , UUID.randomUUID().toString()
                    , MessageType.HEART_BEAT_REQUEST.getFlag());
            Message message = new HeartbeatMessage(header , heartbeat.toByteArray());
            client.writeAndFlush(message);
        }, 2);
//        ChannelFuture closeFuture = client.closeFuture();
//        logger.info("{}" , "close client");
//        closeFuture.sync();
//        group.shutdownGracefully();
    }
}
